package com.exsun.gateway.jt809.kafka;

import com.exsun.gateway.jt809.common.LinkMgr;
import com.exsun.gateway.jt809.common.LinkType;
import com.exsun.gateway.jt809.data.PushData;
import com.exsun.gateway.jt809.handler.SendMgrHandler;
import com.google.gson.reflect.TypeToken;
import exsun.jt808.data.config.CommonMsgTopic;
import exsun.jt808.data.config.jt808.Jt808ReceiveMsgTopic;
import exsun.jt808.data.config.jt809.Jt809MsgIdConfig;
import exsun.jt808.data.config.jt809.Jt809ReciveMsgTopic;
import exsun.jt808.data.jt809_data.BasicVehicleInfo;
import exsun.jt808.data.jt809_data.PushJT809Data;
import exsun.jt808.data.jt809_data.SendBaseEntity;
import exsun.jt808.data.jt809_data.VehicleRegistrationInformation;
import exsun.jt808.data.jt809_data.alarm.main.UpWarnMsgInfoEX;
import exsun.jt808.data.jt809_data.alarm.main.UpWarnMsgOperationInfoEx;
import exsun.jt808.data.jt809_data.vehicledynamic.main.UpBaseMsgVehicleVerify;
import exsun.jt808.data.jt809_data.vehicledynamic.main.UpExgMsgSafetyTerminal;
import exsun.jt808.data.jt809_data.video.main.UpAlarmMediaList;
import exsun.jt808.data.upload_data.GpsData;
import exsun.jt808.data.upload_data.jt808.AlarmDetail;
import exsun.jt808.data.util.GsonUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;


import java.lang.reflect.Type;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

@Service
public class JT809KafkaPushDataToPlatformHandler {

    private Logger logger = LoggerFactory.getLogger(JT809KafkaPushDataToPlatformHandler.class);

    @Autowired
    private SendMgrHandler sendMgrHandler;

    SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");

    //从Kafka监听809消息
    @KafkaListener(topics = CommonMsgTopic.distributeToJT809, containerFactory = "PushDataToPlatform809GroupFactory")
    public void batchPushDataToPlatformConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        try {
            List<PushData> temp = done(records);
            LinkMgr.push2FitChannel(temp, LinkType.MASTER);
            //发送错误，则不提交偏移量
            if(temp == null)return;
//                if(pushs.size() != 0){
//                    logger.info("上报注册：{}条", pushs.size());
//                }
            ack.acknowledge();
        }catch(Exception e){
            logger.info(e.getMessage());
        }
    }

    public List<PushData> done(List<ConsumerRecord<?, ?>> records) {
        List<PushData> temp = new ArrayList<>();
        for (ConsumerRecord<?, ?> record : records) {
//            //任意链路都不通，则不处理数据
//            if (!LinkMgr.getLinkStatus()) return null;
            String msg = record.value().toString();
            SendBaseEntity object = GsonUtils.getObjectFromJson(msg, SendBaseEntity.class);
            switch (object.msgId){
                case Jt809MsgIdConfig.UP_EXG_MSG_SAFETY_TERMINAL:
                    UpExgMsgSafetyTerminal upExgMsgSafetyTerminal = new UpExgMsgSafetyTerminal();
                    upExgMsgSafetyTerminal = GsonUtils.getObjectFromJson(msg,UpExgMsgSafetyTerminal.class);
                    upExgMsgSafetyTerminal.distributeTime = new Date();
//                    upExgMsgSafetyTerminal.install_time = new Date();
                    byte[] body = sendMgrHandler.jt809Encode.UpExgMsgSafetyTerminal(upExgMsgSafetyTerminal,object.msgId);
                    PushData pushData = new PushData(Jt809MsgIdConfig.UP_EXG_MSG_SAFETY_TERMINAL, body);
                    temp.add(pushData);
                    break;
                case Jt809MsgIdConfig.UP_WARN_MSG_INFO_EX:
                    UpWarnMsgInfoEX upWarnMsgInfoEX = new UpWarnMsgInfoEX();
                    upWarnMsgInfoEX = GsonUtils.getObjectFromJson(msg,UpWarnMsgInfoEX.class);
                    upWarnMsgInfoEX.distributeTime = new Date();
                    byte[] body2 = sendMgrHandler.jt809Encode.UpWarnMsgInfoEX(upWarnMsgInfoEX,object.msgId);
                    PushData pushData2 = new PushData(Jt809MsgIdConfig.UP_WARN_MSG_INFO_EX, body2);
                    temp.add(pushData2);
                    break;
                case Jt809MsgIdConfig.UP_WARN_MSG_OPERATION_INFO_EX:
                    UpWarnMsgOperationInfoEx upWarnMsgOperationInfoEx = new UpWarnMsgOperationInfoEx();
                    upWarnMsgOperationInfoEx = GsonUtils.getObjectFromJson(msg,UpWarnMsgOperationInfoEx.class);
                    upWarnMsgOperationInfoEx.distributeTime = new Date();
                    byte[] body3 = sendMgrHandler.jt809Encode.UpWarnMsgOperationInfoEx(upWarnMsgOperationInfoEx,object.msgId);
                    PushData pushData3 = new PushData(Jt809MsgIdConfig.UP_WARN_MSG_OPERATION_INFO_EX, body3);
                    temp.add(pushData3);
                    break;
                case Jt809MsgIdConfig.UP_WARN_MSG_FILELIST_REQ:
                    UpAlarmMediaList upAlarmMediaList = new UpAlarmMediaList();
                    upAlarmMediaList = GsonUtils.getObjectFromJson(msg,UpAlarmMediaList.class);
                    upAlarmMediaList.distributeTime = new Date();
                    byte[] body4 = sendMgrHandler.jt809Encode.upAlarmMediaList(upAlarmMediaList,object.msgId);
                    PushData pushData4 = new PushData(Jt809MsgIdConfig.UP_WARN_MSG_FILELIST_REQ,body4);
                    temp.add(pushData4);
                    break;
            }
        }
        return temp;
    }



}
